package com.shujia.profile.behavior.commodity

import com.shujia.profile.UserProfile
import com.shujia.util.SparkUtil
import org.apache.spark.sql.{DataFrame, SparkSession}

class UserProfileFavGoods extends UserProfile {
  flagName = "favGoods"

  /**
   * 用于提取用户画像的接口
   *
   * @param spark ： spark环境
   * @return
   */
  override def extractProfile(spark: SparkSession): DataFrame = {

    /**
     * collect_set(去重),collect_list(不去重)：组内将某一个列聚合成一个数组，和explode功能相反
     * concat_ws,通过一个分隔符将一个数组拼接成一个字符串，和split功能相反
     */

    spark.sql(
      s"""
         |select
         |member_id as id,
         |concat_ws(',',collect_set(commodity_id)) as favGoods
         |from
         |(
         |    select * from dwd.dwd_t_order_d_f
         |    where ds='$ds'
         |
         |) as a
         |join
         |(
         |   select * from dwd.dwd_t_order_commodity_d_f
         |   where ds='$ds'
         |)
         | as b
         | on a.order_id = b.order_id
         | group by member_id
         |""".stripMargin)
  }
}

object UserProfileFavGoods {
  def main(args: Array[String]): Unit = {
    //获取参数,在提交任务时，在包名后面增加的参数
    val ds: String = args.head
    //spark环境
    val spark: SparkSession = SparkUtil.getSession()

    new UserProfileFavGoods().run(spark, ds)
  }
}
